package com.felk.kafka.java;

import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.Config;
import org.apache.kafka.clients.admin.CreateTopicsResult;
import org.apache.kafka.clients.admin.DeleteTopicsResult;
import org.apache.kafka.clients.admin.DescribeConfigsResult;
import org.apache.kafka.clients.admin.DescribeTopicsResult;
import org.apache.kafka.clients.admin.ListTopicsOptions;
import org.apache.kafka.clients.admin.ListTopicsResult;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.admin.TopicDescription;
import org.apache.kafka.clients.admin.TopicListing;
import org.apache.kafka.common.config.ConfigResource;

import java.util.Arrays;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;

/**
 * @Author alison
 * @Date 2024/4/10 16:47
 * @Version 1.0
 * @Description
 */
public class E3_Admin {


    private final static String brokers = "192.168.56.101:9092";

    private AdminClient adminClient;

    private final static String default_topic = "test_topic";

    public static void main(String[] args) {

    }

    public static AdminClient adminClient() {
        return createAdminClientByProperties();
    }

    public static AdminClient createAdminClientByProperties() {
        Properties prop = new Properties();
        // 配置Kafka服务的访问地址及端口号
        prop.setProperty(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, brokers);
        // 创建AdminClient实例
        return AdminClient.create(prop);
    }

    /**
     * 创建AdminClient的第二种方式
     */
    public static AdminClient createAdminClientByMap() {
        Map<String, Object> conf = Maps.newHashMap();
        // 配置Kafka服务的访问地址及端口号
        conf.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, brokers);
        // 创建AdminClient实例
        return AdminClient.create(conf);
    }

    /**
     * 创建Topic实例
     */
    public static void createTopic(String topic) {
        AdminClient adminClient = adminClient();
        //副本因子
        Short re = 1;
        NewTopic newTopic = new NewTopic(topic, 1, re);
        CreateTopicsResult createTopicsResult = adminClient.createTopics(Arrays.asList(newTopic));
        System.out.println("CreateTopicsResult : " + createTopicsResult);
        adminClient.close();
    }

    /**
     * 获取topic列表
     */
    public static void topicList() throws Exception {
        AdminClient adminClient = adminClient();
        //是否查看Internal选项
        ListTopicsOptions options = new ListTopicsOptions();
        options.listInternal(true);
        //ListTopicsResult listTopicsResult = adminClient.listTopics();
        ListTopicsResult listTopicsResult = adminClient.listTopics(options);
        Set<String> names = listTopicsResult.names().get();
        //打印names
        names.stream().forEach(System.out::println);

        Collection<TopicListing> topicListings = listTopicsResult.listings().get();
        //打印TopicListing
        topicListings.stream().forEach((topicList) -> {
            System.out.println(topicList.toString());
        });
        adminClient.close();
    }

    /**
     * 删除topic
     */
    public static void delTopic(String topic) throws Exception {
        AdminClient adminClient = adminClient();
        DeleteTopicsResult deleteTopicsResult = adminClient.deleteTopics(Arrays.asList(topic));
        deleteTopicsResult.all().get();
    }

    /**
     * 获取topic的描述信息
     */
    public static void describeTopics(List<String> topics) throws Exception {
        // 创建AdminClient客户端对象
        AdminClient adminClient = adminClient();
        // 获取Topic的描述信息
        DescribeTopicsResult result = adminClient.describeTopics(topics);
        // 解析描述信息结果, Map<String, TopicDescription> ==> topicName:topicDescription
        Map<String, TopicDescription> topicDescriptionMap = result.all().get();
        topicDescriptionMap.forEach((topicName, description) -> System.out.printf("topic name = %s, desc = %s \n", topicName, description));
        // 关闭资源
        adminClient.close();
    }

    /**
     * 获取topic的配置信息
     */
    public static void describeConfigTopics(List<String> topicNames) throws Exception {
        // 创建AdminClient客户端对象
        AdminClient adminClient = adminClient();
        List<ConfigResource> configResources = Lists.newArrayListWithCapacity(64);
        topicNames.forEach(topicName -> configResources.add(
                // 指定要获取的源
                new ConfigResource(ConfigResource.Type.TOPIC, topicName)));

        // 获取topic的配置信息
        DescribeConfigsResult result = adminClient.describeConfigs(configResources);

        // 解析topic的配置信息
        Map<ConfigResource, Config> resourceConfigMap = result.all().get();
        resourceConfigMap.forEach((configResource, config) -> System.out.printf("topic config ConfigResource = %s, Config = %s \n", configResource, config));

        // 关闭资源
        adminClient.close();
    }

}
